package org.myfram.flink.flinkonjar.common.connector.kafka.source;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Properties;

public class KafkaSourceBuilder {

    public static DataStreamSource<String> build(StreamExecutionEnvironment env, String topic, Properties props) {
        org.apache.flink.connector.kafka.source.KafkaSourceBuilder<String> builder = KafkaSource.builder();
        builder.setTopics(topic);
        builder.setProperties(props);
        builder.setValueOnlyDeserializer(new SimpleStringSchema());
        return env.fromSource(builder.build(), WatermarkStrategy.noWatermarks(), "KafkaSource");

    }

}
